[レポート] Spotify社のコンテンツ分析チームがBigQueryのデータ消化不良をdbtで回避した方法 #dbtCoalesce

大阪オフィスの玉井です。

2022年10月17日〜21日に行われたCoalesce 2022というハイブリッド（オンライン+オフライン）カンファレンスが開催されました。主催はdbt labs社です。

本記事は、その中で発表されたHow the Content Analytics team at Spotify avoids data indigestion in BigQuery with dbtというセッションについて、レポートをお届け致します。

セッション概要

登壇者

• Nick Baker
  • Senior Analytics Engineer , Spotify
• Brian Pei
  • Analytics Engineer, Spotify

超概要

超有名サブスクリプションサービスであるSpotify社のデータ分析環境に関する話です。尋常ではない大容量データをスムーズに自分の環境に取り込めるように社内開発したWaluigiというツールの話がメインです。

セッションレポート

前段

皆さん、お集まりいただき、本当にありがとうございます。このような素晴らしい講演者の方々と私たちの講演をご覧いただき、本当にありがとうございます。また、dbt labsの皆様、主催していただきありがとうございました。

今回は、弊社のコンテンツ分析チームがデータの消化不良を回避する方法した方法についてお話します。

では、私たちが何者で何をやっているのか、その背景を少しご紹介しましょう。

Nickと私は、Spotifyのコンテンツ戦略（Strategy）・洞察（Insight）・分析（Analytics）チーム（略称 SIA）に所属しています。私たちのチームは、データサイエンティスト、ユーザーリサーチャー、そしてもちろんアナリティクスエンジニアで構成されており、プラットフォーム上でのコンテンツ消費に関するデータや、それが会社として（Spotifyに）どのような影響を与えるかについて研究しています。このチームは6年前からあり、アナリティクスエンジニアリングは3年前から取り組んでいます。

私たちはdbtを採用しています。私が入社してすぐに、データパイプラインと複雑なDAGを支援するためにdbtを採用しました。そして、Spotifyのユーザーがクラウド上で作成したデータを、私たちに提供してくれることを期待しています。

私たちのチームは、私自身と、チームを管理しているMitchellとで構成されています。これからたくさん話を聞くことになるのは、最前列のチャットチャンピオン、Brian Pei、そして最前列に座っているSydney、さらに、ここには写っていませんが、育児休暇中のアソシエイトアナリティクスエンジニアもいます。そしておそらくこのトークを見てくれているであろう、チームのトップであるTimです。Tim、ぜひ見ておいてくださいね。そして、（この登壇で）私たちがあなたの誇りになることを願っています。

今日は、NickがSpotifyのアナリティクスエンジニアリングについて、私たちがdbtを採用したときの様子を含めて説明します。そして、私たちが直面した課題と、私たちの採用をシームレスにし、潮流に乗せるために考え出したソリューションのいくつかを紹介します。

また、Waluigiについてお話します。これは私たちの内部用ユーティリティパッケージで、私たちの仕事に大いに役立っているものです。

Spotify社のデータ分析環境

Spotifyは多くの方がご存知だと思っています。手始めに、やり過ぎない程度に、ちょっとだけエンゲージしてみましょう。Spotifyを使ったことがある方、手を挙げてください。

（おそらく、たくさんの方が挙手）

素晴らしい。ここ1ヶ月以内にSpotifyを使ったことがある人は、どんどん手を挙げてください。

（おそらく、たくさんの方が挙手）

素晴らしい。先週使った人は？昨日使った人は？

（おそらく、たくさんの方が挙手）

すごくグレートです。

この部屋にいるSpotifyを使っている人たちを想像してみてください。そして、それを何億人もの月間アクティブ・ユーザーに当てはめて、会社として、あなたが毎日どんな曲をストリーミングで聴いているか考えてみてください。これらのデータは、私たちが毎日見ているデータであり、特にコンテンツに関して、私たちが行っている仕事の多くを占めています。私たちは、毎日何十億行、何テラバイトものデータを見ていることになります。私たちがここで話していることはすべて、膨大な量のデータを扱っているということです。

では、そのデータをどうするのか？Spotifyでは、2万以上のバッチデータパイプラインを1000のリポジトリに定義し、毎日300のチームによって管理されています。これは大変な数です。つまり、Spotifyには中央集権的なデータチームがなく、データ組織は分散しています。これは、さまざまな状況に柔軟に対応し、迅速に行動できるという意味で素晴らしいことです。

しかし、もちろん、ご想像の通り、チーム間やプロジェクト間の依存関係がある場合は、さらなるオーバーヘッドが発生します。そのような場合、どのように対処すればよいのでしょうか。

私たちは、Google BigQuery及びGoogle Cloudのデータ分析系サービスを5、6年間使っています。つまり、BigQueryは私たちが最もよく使うデータウェアハウスなのです。

Spotifyのアナリティクスエンジニアは、これまでLuigiの上に構築された、Styxというオーケストレーションされた内部変換ツールを利用してきました。このツールは、dbtが実行できることのほとんどを行い、BigQueryとうまく連携しています。また、オープンソースの開発者ポータルであるBackstageとも統合されています。

Styxについてご存じない方のために説明します。これは基本的にSpotify社内でスケジュールやオーケストレーションを扱うツールです。このツールは、あるタイミングが来たらKubernetes上でDockerイメージをスピンアップしているんです。そして、私たちが実行したいワークフローを実行します。これが処理、スケジューリング、その他もろもろです。

それからLuigiですが、これは元々Spotifyが開発したPythonパッケージで、基本的にはDAGを構築するだけですが、ワークフローを個別のタスクとして定義し、依存関係を持たせることができます。dbtを使用している人は皆、DAGに非常に精通しています。Luigiは、ずっと前に開発されたものですが、非常に異なる方法です。

そして最後に、Backstageです。Backstageは、オープンソースの開発者ポータルサイトです。本日の講演では、データエコシステムを可視化するための有用性に焦点を当てたいと思います。しかし、Backstageが行うクールな機能には、今回説明できないものがたくさんあります。

データですが、内部で生成されたSpotifyからのデータもありますし、外部からのデータもあります。そして、何らかの方法でそれをBigQueryに取り込みます。現在、私たちが力を入れているのは、この点です。先ほどお話したようなさまざまなプロジェクトでは、BigQueryに取り込まれたデータを内部で変換するツールで変換した後、再びBigQueryに戻しています。もちろん、これらすべての異なるプロジェクト間で相互依存関係が発生します。これらをオーケストレーション用のStyxで包み込み、Backstageから見えるようにしています。

では、Backstageを簡単にお見せしましょう。これはワークフローの非常にシンプルな例で、Backstageには、単一のモデルまたは複数のモデルが一緒に動作しており、一般的な情報を持っています。また、ヘルスレポートやスケジュールされたジョブのステータスなど、ドキュメントも用意されています。これは、変換を実行するときに表示されるものです。これはSpotifyのワークフローを概観するためのものです。

もう少し掘り下げると、インスタンスの例もあります。この例では、2020年4月7日に1つの日次ジョブをスケジュールしている場合、そのジョブが実行されようとしている様子を見ることができます。この例ではパーミッションのエラーがたくさん出ていて、いつもイライラさせられますが、右側のログを調べれば、それに対応するGitのコミットがわかります。最終的に、この実行は成功しました。

このように、私たちはさまざまなワークフローにモニタリングを組み込んでいます。

また、Backfillを実行するのにも最適です。Backfillを実行する必要がある場合、データ変換を一度に一日ずつ実行するように設定しています。Backfillを実行するためのスケジュールを組むには、Backstageを使用します。

そして最後に、非常に重要なことですが、ワークフローの上流と下流の依存関係を可視化することができます。これによって、私たちのワークフローが何に依存しているのか、またどのワークフローが私たちのワークフローに依存しているのかを理解することができますし、何か問題があれば、人々とコミュニケーションを取ることができるようになり、非常に便利なツールです。しかし、繰り返しになりますが、これは私たちのデータエコシステム全体に対する一種のユーザーインターフェースに過ぎません。

Spotify社とdbt

さて、ここからは、dbtのお話と、現在の状況をお伝えします。

現在でも、社内のほとんどの人が、これまで標準的に使われてきた社内ツールを使っていますし、これからもそれは標準的なツールであり続けるでしょう。しかし、dbtは、私たちのチームから始まり、今では他のいくつかのチームにも広がり、本番環境でも使用することができるようになりました。

まず第一に、私は皆さんにdbtを売り込む必要はありません、それは馬鹿げてるでしょう。私たちは皆、dbtに賛同していますし、素晴らしいことだと思います。

私たちには課題…長いクエリやレイヤーの欠如等がありました。内部の変換ツールを見ると、サブクエリして〜またサブクエリして〜それをまたサブクエリで〜という具合に、維持するのが不可能な状態になっています。

また、この内部ツールでは、本番用と開発用のリポジトリを別々に用意する必要があったことも重要です。そのため、メンテナンス担当者が対応しなければならないオーバーヘッドが増えるだけです。そして、それは不透明なものです。Spotifyの平均的なアナリティクスエンジニアは、そのツールに貢献することができません。

私たちは皆、自分の仕事をもっと効率的にしたい、もっと簡単にしたい、あまり同じことを繰り返したくないと思っています。そのうえで、さらにいくつかの課題があります。先ほども申し上げたように、弊社には大量のデータがあります。dbtとBackstageを統合し、UIを使用してジョブ実行を調整し、スケジュール化する方法を見つける必要がありました。

また、この講演の残りの大部分を費やすことになりますが、ミックスドテーブルの扱いについても説明します。旧来のBigQueryのようなものについてお話します。そして最後に、とても重要なことですが、別々のdbtプロジェクトにまたがる依存関係の管理についてです。

Spotifyが開発したデータ分析のためのツールなど

私たちのチームは当然ながら試行錯誤を繰り返し、自分たちでこれらのことをすべて把握するため、dbt用のDockerコンテナを作成し、そして、それを他のチームも利用できるようにする必要がありました。

そこで、他のチームがForkできるようなテンプレートを作成し、基本的にはBigQueryに接続するためのプリベイクドガイダンスが用意されています。また、CI/CDやすべてのBackstageインテグレーションを確認することもできます。

重要なのは、実行ごとに生成されるマニフェストJSONを解析する機能です。これにより、dbtプロジェクトやテーブルからワークフローの異なるエンドポイントを接続・登録し、それを実行するワークフローをBackstageに登録して、必要な可視性を提供することができます。また、Backstageで見ることのできるドキュメントや、他のチームにとっても重要なリネージ情報を提供することもできます。

さらに、dbt_run() command generatorも注目すべき機能です。Backstageで見たように、Backfillをこの日に実行したい、この日だけ実行するようにスケジューリングしたい、などの要望をDockerイメージに送り込み、定義したワークフローに沿ったものだけを実行できるようにしなければなりません。そのため、dbtを1回だけ実行するのではなく、タグを使用して、実行するたびに使用するモデルを選択することがよくあります。

私たちは、dbtのユーティリティパッケージであるWaluigiを作りました。私たちは明らかにデザインは苦手です。Luigiのロゴを切り取って、Googleスライドのワードアートを追加しただけです。しかし、これはWaluigiで、Spotify社内で作られたオリジナルのLuigiツールへのオマージュですが、私たちはそれを少し反転させたのです。

これらの部品がどのように組み合わされているかを視覚化するために、このような図を作成しました。

まず最初に、この非常に単純化された図をご覧ください。基本的には、コンテナテンプレートの役割を果たすベースイメージがあり、dbt coreをインストールするために必要な手順と、それに対応する重要なBackstageインテグレーションがあります。

そして、Waluigiです。これはパッケージで、マクロ、テスト、バージョン管理されたdbtプロジェクトです。これは、Artifactoryを通じて個別にメンテナンスされています。Spotifyの内部パッケージのメンテナンスに使っています。これは、他のチームの開発方法と一致させるためです。特に、自分たちの内部パッケージの管理は、他のチームと合わせるのが好きなんです。

Waluigi Classicパッケージとは、会社全体で使う共通のマクロで、dbtを導入する人が簡単に先に進んで、その過程で学んだ教訓を活用できるようにするものです。これはテストに関するもので、ユーザー定義関数や一般的なSQLジェネレーターが含まれています。

そして重要なのは、統合テストも含まれていることです。つまり、出力されるマクロがすべて期待通りのものであることを確認するために、マクロを作成する際の仮定をテストしているのです。

マニフェストJSONの出力は、Backstageでの可視性、ドキュメントの登録、依存関係などを確認することをを想定しています。これらはすべて、Waluigiの中でテストされます。

そして最後に、この話の残りの部分をこれに費やすのですが、get_external_table()と呼ばれるものです。これは、外部の依存関係を処理するために使うマクロのコレクションです。

「外部依存」なデータの取り扱いについて

さて、それでは「外部依存」についてです。これは、外部にあるデータではなく、また、BigQueryの外部にあるデータをBigQueryに取り込むのでもなく、BigQueryにすでにあるデータを取り込むということです。

どういうデータかというと、BigQueryにすでにあるデータで、おそらくデータパイプライン上流の他のチームによって変換されたものです。dbtを使ったジョブを実行するために、上流のモデルをすべて取り込む必要があります。

ではまず、BigQueryの歴史について少しお話します。

Spotifyでは、テーブルの構造が混在しています。

1つ目はテーブルシャーディングです。私たちは長い間、ちょっとクレイジーなユーザーだったので、2018年以前はBigQueryでテーブルを保存する唯一の方法は、テーブルシャーディングを介するものでした。テーブルシャーディングとは、ネーミングプレフィックスを使って複数のテーブルにデータを格納することです。

この例では、私のプールテーブルがありますが、これはこのテーブルのネーミングプレフィックスです。そしてもちろん、それぞれの日付はサフィックスで表現されています。2026年のデータにアクセスしたい場合は、そのテーブルから、あるいは2720年のデータにアクセスしたい場合は、そのテーブルから選択します。つまり、2026年のデータにアクセスしたい場合は、2026年のテーブルから、2720年のデータにアクセスしたい場合は2720年のテーブルから、というように、まるで独立したテーブルのように扱われるのです。私たちは、長い間、BigQueryを使用してきたため、私たちが持つデータの大半はそのような構造を持っています。また、社内の変換ツールはそれ以前に開発されたものです。このツールは、そのフォーマットでデータを取り込み、同じフォーマットで出力するように設計されています。ですから、データの大半はこのような構造になっています。

ネイティブパーティショニングですが、私がSpotifyで仕事を始める前に知っていたのはこっちだけでした。基本的には参照するだけなので、ネイティブにパーティショニングされたテーブルからデータにアクセスする場合は、WHERE句を使います。パーティショニングはバックグラウンドで行われています。BigQueryではこの方法がGoogleの推奨する一般的な方法となっています。

さて、魔法の杖を使ってデータウェアハウス全体を変換したいところです。しかし、他のユースケースのために複雑な依存関係があり、また社内の変換ツールもまだ主要なものであるため、私たちの能力をはるかに超えた、ばかげた話になってしまいます。その結果、Googleはこの方法論を採用し、この構造を採用することを推奨しています。

そこで私たちは、dbtにこのフォーマットでテーブルを出力させることにしました。つまり、Spotifyのテーブルの大半は、スライドの上のようにシャード化されています。そしてdbtで読み込んでいるテーブルは、一部を除いてネイティブにパーティショニングされたテーブルとして書き出されます。つまり重要なのは、エコシステムが混在しているということです。

また、そのモデルをどのように実行するかも呼び出したいところです。この例では、Backstageで、2020年4月7日のデータを扱いたい、というケースです。dbtのrunコマンドには、partition_run_idという変数があります。つまり、このインスタンスで実行するのは何日のデータなのか？ということですね。これは、扱うべきデータが多すぎるという事実をコントロールするためのものです。

そうすると、そのデータを実行したいときに、確かに1日分のデータでモデルを実行したいと思います。しかし、上流で1日分のデータしかアクセスしたくないというわけではありません。場合によっては、テーブル上のすべての日付のパーティションをプルインする必要があります。また、3日分の集計を行う場合もあります。そのため、取り込みたいパーティションや更新をリストアップする必要があります。

また、古典的なケースとして、今実行しているパーティションと同じデータを上流から取得することもあります。また、メタデータなど、最新のデータにアクセスしたい場合は、時間が経つにつれてデータの精度が上がっていることを前提に運用します。ですから、過去にさかのぼってアクセスする必要はないのです。

全てのパーティションにアクセスしたい場合、ここで行うことは、SELECT文の中で、Waruigiからget_external_tableを呼び出し、テーブル名をexternal shartedに設定して、そのテーブルの形状を示します。

そしてオペレーションを設定します。今回は、all_partitionsを使用していますが、これは単純にSELECT * FROM my_cool_tableの最後にアスタリスクをつけたものとなります。このアスタリスクはBigQueryではワイルドカードとして使用できます。つまり、この接頭辞の後に続くもの全てのデータを与えてくれるということです。これでMERGEがうまくいき、満足です。

さて、最も一般的な使用例では、一度に1日だけ実行することになります。その場合、partition_run_idを使用します。先程と同じように、外部テーブルを取得し、テーブル名を設定し、partition_run_idというオペレーションを選択します。これをコンパイルすると、テーブル名の末尾にpartition_run_idが追加されるだけです。何らかの理由でデータが欠落している場合は、エラーになります。もしデータがあれば、素晴らしいことに、期待通りに実行され、テーブルにMERGEされます。

これらのケースに対応できるようなオペレーションを構築しました。

基本的にアナリティクスエンジニアは、プロジェクトの上流にある何百ものテーブルを参照しながら、「この特定のデータだけを取り込みたいんだけど、1日分とか3日分とか」と言うことがほとんどです。というのも、私たちはストリームだけでも数十億行、数テラバイトのデータを毎日扱っているのですから、その他のデータについても同様です。Spotifyでは、BigQueryでデータの大部分をカバーしています。

私たちはdbtを使って、ネイティブパーティショニングされたテーブルから書き出しているのです。そして、みんなにdbtを使ってもらいたいんです。なぜなら、私たちはdbtが大好きだからです。より良いツールですからね。

ということは、dbtのプロジェクトには、ネイティブパーティショニングされたテーブルを出力する、複雑な依存関係が存在することになります。そのため、外部テーブルを適応させる必要があります。

また、同じユースケースでも、形が少し違うだけで、all_partitons、partition_list、partition_run_id、 most_recent、そしてもちろん、ここでは紹介していないその他の形式を使う必要がありますね。

なぜ外部データにsourceを使わないのか不思議に思っている方もいらっしゃると思いますが、これにはいくつかの理由があります。

ひとつは、私たちには何百もの上流依存関係があり、そのためにYAMLファイルを維持するのは面倒で、ちょっと残念なことになります。

また、外部テーブルを取得することで、キーとなる情報を設定に読み込み、最終的にマニフェストJSONとしてBackstageに取り込みます。この情報は基本的に、dbtの実行ごとに、どのテーブルが読み込まれたか、どのように読み込まれたか、どのパーティションが使用されたかを教えてくれます。最終的には、その操作を選択するためのマクロが必要でした。そのため、データに対してどの戦略を選択するかを動的に選択できるようなものを開発する必要があったのです。しかし、そうすると、問題が発生します。

この例では、sourceと書いています。これはsourceの正しい構文ではありませんが、このように書いています。そして、dateがこのマクロパーティションの実行日に等しいと言うWHERE句があるだけですね。partition_run_idを日付に変換して、コンパイルして実行するだけです。コンパイルされたSQLは、なかなかいい感じでしょう？私のテーブルから、日付が2022-09-26に等しいものを選びました。

データがあれば、このモデルは問題なく実行できます。しかし一方で、当然ながら、そのテーブルはまだ存在していない（書き込まれていない）かもしれません。

問題はこのSQLが完全に有効であることです。もしこのSQLを実行してもエラーにはならず、ただゼロ行を処理し、下流のテーブルのゼロ行をマージすることになります（サイレントエラー）。

もちろん、そのためのテストを設定することは可能です。しかし、これは外部にバンドルしてしまう方が簡単です。それから、さっき言ったように、dbtプロジェクトがネイティブパーティショニングされたテーブルを書き出すと、このようなことが常に起こるようになります。これは非常に危険です。

そこで、他のチームが「レガシーなTransformツールでも扱えるよ」と言わずに、もっと簡単に導入できるようにしたいと考えました。そこでもちろん、get_external_tableに戻ることにします。

先程と構成が少し変わっているのがわかると思います。partition_run_idというオペレーションを設定します。つまり、9月26日、同じ日付のmy_cool_tableのデータにのみアクセスする、ということです。

これを実行すると、まずBigQueryのインフォメーションスキーマをチェックすることになります。パーティションリストや、すでに書き込まれた日付のリストを引き出します。そして、それを（先程設定した）operationの値と照らし合わせてみます。partition_run_idに対して実行する場合、この日付が有効な日付のリストにあるかどうか、もしチェックする日付の範囲が広ければ、それを繰り返し、リストに対してすべての日付をチェックします。直近の日付であれば、過去にさかのぼってチェックします。そして、どのパーティションがすでに書き込まれたかを調べ、最も新しいものを取り出し、それをモデルにコンパイルします。もし必要な日付がまだ書き込まれていないことがわかったら、BigQueryはエラーになります。もしモデルがデータがそこにあればOKです。それは私たちが望むように正確に動作します。

このように、さまざまなオペレーションで、さまざまなバージョンのパーティションテーブルをネイティブに扱えるようにしました。

まとめ

Waluigiのget_external_tableを要約してみましょう。

まず最初に、dbtプロジェクトの外部にあるBigQueryのデータにアクセスします。テーブル構造がどうであれ（シャーディングされていようが、ネイティブパーティションであろうが）必要なデータを選択することができます。また、そもそも上流のデータが存在するかどうかのチェックも組み込まれています。サイレントエラーは避けたいです。テーブルが空のテーブルを書き込んだかを調べるために時間を浪費するようなことは辛いですからね。

それから、今回はあまり詳しく触れませんが、将来的にはマニフェストJSONに何を書き出すかについても触れます。

このように、私たちはすべてを統合することができます。Backstage、依存関係、ドキュメント、必要なものすべてです。そして、dbtのプロジェクトを相互に依存させることができるようになったというのが、ここでの重要な点です。

だから、簡単に言うと、Spotifyでは、Waluigiがいなかったら、たくさんの無駄な作業を課せられることになるんです。

Spotifyには、50を超えるdbtプロジェクトがあります。もちろん、そこには多くのPoCがあり、好奇心旺盛なエンジニアやアナリティクスエンジニアなど、このプロジェクトで何ができるかを試している人たちがいます。

しかし、10～15個のプロジェクトが本番用のテーブルを作成し、それらのプロジェクト間で依存関係があることは、自信を持って断言できます。ですから、将来を見据えて、何を開発するかについてもう少し熟慮する必要があります。

これまでは、すでにあるツールに対応するものが多かったのですが、これからは新しい要素も加えていきます。ですから、ロードマップを作成し、より具体的に開発を進めています。

また、Spotifyの他のアナリティクスエンジニアがWaluigiにどのように貢献できるのか、そのプロセスもまだ確立されていません。そのためのワークフローを確立する必要があります。

もうひとつは、Backstageをどのように統合したかについて、より良いドキュメントを作成することです。アナリティクスエンジニアが、そのような作業を手伝ったり、世界に公開したりするのに役立つと思います。そして最終的なゴールは、WaluigiがSpotifyの公式データ変換ツールになり、dbtをモダンデータスタックの重要な部分として位置づけることです。ということで、私たちの発表はこれで終わりです。

おわりに

規模がデカすぎて、上流データを取り込むのにこんなに苦労するのか…と思いました。技術力のある企業は、dbtを使うだけでなく、dbt用のパッケージも内製してしまうのがすごいですね。

あとは、Waluigiのロゴ画像は、法的に大丈夫なのか、心配です。